package com.tiancheng.trade.order.mq.config;


import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.tiancheng.trade.commom.core.enums.TradeRabBitMqEnums;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.retry.interceptor.RetryOperationsInterceptor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.Resource;


/**
 * @author jiangjun
 * @description 消息队列相关配置
 * @date 2024/10/14 23:43
 */
@Slf4j
@Getter
@Configuration
public class RabbitMqConfig {

    @Bean
    public MessageConverter jsonMessageConverter() {
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());

        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
        return new Jackson2JsonMessageConverter(objectMapper);
    }

    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, ConnectionFactory connectionFactory, ThreadPoolTaskExecutor taskExecutor) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        configurer.configure(factory, connectionFactory);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(10);
        factory.setTaskExecutor(taskExecutor);
        return factory;
    }


    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jsonMessageConverter());
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                log.error("消息发送失败: {}", cause);
            }
        });
        rabbitTemplate.setReturnsCallback(returned -> {
            log.error("消息无法路由: exchange={}, routingKey={}, replyCode={}",
                    returned.getExchange(), returned.getRoutingKey(),
                    returned.getReplyCode());
        });
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }

    // -----------------------------------------------------------------支付中心推送订单中心支付消息

    /**
     * 支付消息通知实际消费队列所绑定的交换机
     */
    @Bean
    DirectExchange paymentStatusDirect() {
        return (DirectExchange) ExchangeBuilder.directExchange(TradeRabBitMqEnums.QUEUE_PAYMENT_NOTIFY.getExchange()).durable(true).build();
    }

    /**
     * 支付消息通知迟队列队列所绑定的交换机
     */
    @Bean
    DirectExchange paymentStatusTtlDirect() {
        return (DirectExchange) ExchangeBuilder.directExchange(TradeRabBitMqEnums.QUEUE_TTL_PAYMENT_NOTIFY.getExchange()).durable(true).build();
    }


    /**
     * 订单实际消费队列
     */
    @Bean
    public Queue paymentStatusQueue() {
        return new Queue(TradeRabBitMqEnums.QUEUE_PAYMENT_NOTIFY.getName());
    }

    /**
     * 订单延迟队列（死信队列）
     */
    @Bean
    public Queue paymentStatusTtlQueue() {
        return QueueBuilder.durable(TradeRabBitMqEnums.QUEUE_TTL_PAYMENT_NOTIFY.getName()).withArgument("x-dead-letter-exchange", TradeRabBitMqEnums.QUEUE_PAYMENT_NOTIFY.getExchange())//到期后转发的交换机
                .withArgument("x-dead-letter-routing-key", TradeRabBitMqEnums.QUEUE_PAYMENT_NOTIFY.getRouteKey())//到期后转发的路由键
                .build();
    }

    /**
     * 将订单队列绑定到交换机
     */
    @Bean
    Binding paymentStatusBinding(DirectExchange paymentStatusDirect, Queue paymentStatusQueue) {
        return BindingBuilder.bind(paymentStatusQueue).to(paymentStatusDirect).with(TradeRabBitMqEnums.QUEUE_PAYMENT_NOTIFY.getRouteKey());
    }

    /**
     * 将订单延迟队列绑定到交换机
     */
    @Bean
    Binding paymentStatusTtlBinding(DirectExchange paymentStatusTtlDirect, Queue paymentStatusTtlQueue) {
        return BindingBuilder.bind(paymentStatusTtlQueue).to(paymentStatusTtlDirect).with(TradeRabBitMqEnums.QUEUE_TTL_PAYMENT_NOTIFY.getRouteKey());
    }

    /**
     * 支付结算实际消费队列所绑定的交换机
     */
    @Bean
    DirectExchange paymentSettleDirect() {
        return (DirectExchange) ExchangeBuilder.directExchange(TradeRabBitMqEnums.QUEUE_SETTLE_MESSAGE.getExchange()).durable(true).build();
    }

    /**
     * 支付结算实际消费队列
     */
    @Bean
    public Queue paymentSettleQueue() {
        return new Queue(TradeRabBitMqEnums.QUEUE_SETTLE_MESSAGE.getName());
    }

    /**
     * 将支付结算绑定到交换机
     */
    @Bean
    Binding paymentSettleQueueBinding(DirectExchange paymentSettleDirect, Queue paymentSettleQueue) {
        return BindingBuilder.bind(paymentSettleQueue).to(paymentSettleDirect).with(TradeRabBitMqEnums.QUEUE_SETTLE_MESSAGE.getRouteKey());
    }
}
